package com.mqtt.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.mqtt.handler.ProtocolHandler;
import com.mqtt.service.impl.DeviceSessionService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * API到设备的消息监听器
 * 监听API服务发送的消息，根据type路由到不同的MQTT topic
 */
@Slf4j
@Component
public class ApiToDeviceMessageListener {
    
    @Autowired
    private ProtocolHandler protocolHandler;
    
    @Autowired
    private DeviceSessionService deviceSessionService;
    
    /**
     * 监听API服务发送给设备的消息
     * 消费专门的MQTT命令topic
     */
    @KafkaListener(
        topics = "jianyou_tongxin_MQTT_Command",
        groupId = "${kafka.consumer.group.id}",
        containerFactory = "kafkaListenerContainerFactory"
    )
    public void handleApiToDeviceMessage(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
        try {
            for (ConsumerRecord<String, String> record : records) {
                String watchId = record.key();  // 使用watchId作为key
                String value = record.value();
                
                log.info("Received message from API for device: {}, message: {}", watchId, value);
                
                try {
                    JSONObject message = JSON.parseObject(value);
                    if (message != null && watchId != null && !watchId.isEmpty()) {
                        // 路由消息到对应的MQTT topic
                        routeMessageToDevice(watchId, message);
                    }
                } catch (Exception e) {
                    log.error("Error parsing message for device: {}, error: {}", watchId, e.getMessage());
                }
            }
            
            // 确认消息
            ack.acknowledge();
            
        } catch (Exception e) {
            log.error("Error processing API to device messages", e);
        }
    }
    
    /**
     * 根据消息type路由到不同的MQTT topic
     */
    private void routeMessageToDevice(String watchId, JSONObject message) {
        String type = message.getString("type");
        JSONObject data = message.getJSONObject("data");
        
        log.info("Routing message to device: {}, type: {}", watchId, type);
        
        // 检查设备是否在线
        if (!deviceSessionService.isDeviceOnline(watchId)) {
            log.warn("Device {} is offline, message type: {} will be cached", watchId, type);
            cacheOfflineMessage(watchId, type, message);
            return;
        }
        
        // 根据type确定topic
        String topic = determineTopicByType(watchId, type);
        
        // 构建发送给设备的消息
        Map<String, Object> deviceMessage = buildDeviceMessage(type, data, message);
        
        // 发送到MQTT
        sendToMqttTopic(topic, deviceMessage);
    }
    
    /**
     * 根据type确定MQTT topic
     */
    private String determineTopicByType(String watchId, String type) {
        String baseTopic = "watch/" + watchId + "/";
        
        switch (type) {
            // 命令类消息
            case "command":
            case "control":
                return baseTopic + "cmd";
                
            // 配置类消息
            case "config":
            case "setting":
                return baseTopic + "config";
                
            // 通知类消息
            case "notification":
            case "alert":
                return baseTopic + "notify";
                
            // 聊天消息
            case "chat":
            case "message":
                return baseTopic + "chat";
                
            // 定位请求
            case "location":
            case "position":
                return baseTopic + "location";
                
            // 健康数据
            case "health":
                return baseTopic + "health";
                
            // 闹钟提醒
            case "alarm":
            case "reminder":
                return baseTopic + "alarm";
                
            // 联系人同步
            case "contact":
                return baseTopic + "contact";
                
            // 固件升级
            case "ota":
            case "firmware":
                return baseTopic + "ota";
                
            // SOS紧急
            case "sos":
                return baseTopic + "sos";
                
            // 课程表
            case "schedule":
                return baseTopic + "schedule";
                
            // 电子围栏
            case "fence":
            case "geofence":
                return baseTopic + "fence";
                
            // 默认数据topic
            default:
                return baseTopic + "data";
        }
    }
    
    /**
     * 构建发送给设备的消息
     */
    private Map<String, Object> buildDeviceMessage(String type, JSONObject data, JSONObject originalMessage) {
        Map<String, Object> deviceMessage = new HashMap<>();
        
        // 基础字段
        deviceMessage.put("type", type);
        deviceMessage.put("timestamp", System.currentTimeMillis());
        
        // 如果有消息ID，保留用于追踪
        if (originalMessage.containsKey("msgId")) {
            deviceMessage.put("msgId", originalMessage.getString("msgId"));
        }
        
        // 根据不同类型构建消息内容
        switch (type) {
            case "command":
                deviceMessage.put("cmd", data.getString("cmd"));
                deviceMessage.put("params", data.getJSONObject("params"));
                break;
                
            case "config":
                deviceMessage.put("config", data);
                deviceMessage.put("version", data.getString("version"));
                break;
                
            case "chat":
                deviceMessage.put("from", data.getString("from"));
                deviceMessage.put("content", data.getString("content"));
                deviceMessage.put("contentType", data.getString("contentType")); // text/voice/image
                break;
                
            case "location":
                deviceMessage.put("action", data.getString("action")); // request/response
                deviceMessage.put("interval", data.getIntValue("interval"));
                break;
                
            case "ota":
                deviceMessage.put("version", data.getString("version"));
                deviceMessage.put("url", data.getString("url"));
                deviceMessage.put("md5", data.getString("md5"));
                deviceMessage.put("size", data.getLong("size"));
                deviceMessage.put("force", data.getBooleanValue("force"));
                break;
                
            case "alarm":
                deviceMessage.put("alarmId", data.getString("alarmId"));
                deviceMessage.put("time", data.getString("time"));
                deviceMessage.put("repeat", data.getJSONArray("repeat"));
                deviceMessage.put("label", data.getString("label"));
                deviceMessage.put("enabled", data.getBooleanValue("enabled"));
                break;
                
            case "contact":
                deviceMessage.put("action", data.getString("action")); // add/update/delete/sync
                deviceMessage.put("contacts", data.getJSONArray("contacts"));
                break;
                
            case "fence":
                deviceMessage.put("fenceId", data.getString("fenceId"));
                deviceMessage.put("action", data.getString("action")); // add/update/delete/enable/disable
                deviceMessage.put("shape", data.getString("shape")); // circle/polygon
                deviceMessage.put("coordinates", data.getJSONArray("coordinates"));
                deviceMessage.put("radius", data.getDoubleValue("radius"));
                break;
                
            default:
                // 默认直接放入data
                deviceMessage.put("data", data);
        }
        
        return deviceMessage;
    }
    
    /**
     * 发送消息到MQTT topic
     */
    private void sendToMqttTopic(String topic, Map<String, Object> message) {
        try {
            String payload = JSON.toJSONString(message);
            
            // 根据消息类型决定QoS级别
            int qos = determineQoS(message.get("type").toString());
            
            // 使用ProtocolHandler发送
            protocolHandler.sendMessage(topic, payload, qos, false);
            
            log.info("Message sent to MQTT topic: {}, type: {}", topic, message.get("type"));
            
        } catch (Exception e) {
            log.error("Error sending message to MQTT topic: {}", topic, e);
        }
    }
    
    /**
     * 根据消息类型确定QoS级别
     */
    private int determineQoS(String type) {
        switch (type) {
            case "command":
            case "config":
            case "ota":
            case "sos":
                return 2; // 确保送达一次
            case "alarm":
            case "fence":
            case "contact":
                return 1; // 至少送达一次
            default:
                return 0; // 最多送达一次
        }
    }
    
    /**
     * 缓存离线消息
     */
    private void cacheOfflineMessage(String watchId, String type, JSONObject message) {
        // TODO: 实现离线消息缓存到Redis
        // 当设备上线时，从缓存中取出消息发送
        log.info("Caching offline message for device: {}, type: {}", watchId, type);
    }
}